// -*-mode:c++; coding:utf-8-*-


#include <string>
#include <list>
#include <queue>

#include <xbase_exception.hpp>

#include "naming_strategy.hpp"
#include "global.hpp"
#include "table_schema.hpp"
#include "table_schema_manager.hpp"
#include "memory_table_scanner.hpp"

#include "memory_table.hpp"

namespace xbase {

	// 采用写时拷贝的方式，使得并行性更好，并保证了行操作的原子性
	void memory_table::add(row_block::pointer &new_row_block)
	{
		// 要修改该行块，所以必须持有它的指针才行！另外，其它线
		// 程可能将行块更改了，也必须要用指针才能感知到。
		//
		// 对std::map来说，持有迭代器是安全的，即内部的节点的内
		// 存位置是不会受插入、删除操作影响的，所以也可以安全的
		// 持有指向元素的指针
		
		const std::size_t new_row_bytes = new_row_block->data_bytes();
		
		const row_block::string_type * const row_name = new_row_block->row_name();
		
		boost::upgrade_lock<boost::shared_mutex> u_lock(m_lock);
		const impl_map_type::iterator iter = m_table->find(row_name);
		if(iter != m_table->end()) // 该行已存在，拷贝并替换
		{
			entry_type * const inner_storage = &(iter->second);
			boost::mutex::scoped_lock s_lock(inner_storage->row_mutex()); // 对该行加锁，阻止并发复制写
			// 复制并写入替换
			inner_storage->m_row_block = share_row_block(new_row_block,
								     inner_storage->m_row_block); // 赋值是原子操作，智能指针保证了这一点
		}
		else // upgrade to be write lock and insert into table
		{
			// 更新失败，再尝试写入
			const entry_type entry(share_row_block(new_row_block)); // must make row block shared
			// 限制作用域，避免较长的析构操作时仍持有锁
			{
				boost::upgrade_to_unique_lock<boost::shared_mutex> lock(u_lock);
				const std::pair<impl_map_type::iterator, bool>
					ret = m_table->insert(impl_map_type::value_type(entry.row_name(),
											entry));
				assert(ret.second == true);
				(void)ret;
			}
		}
		u_lock.unlock(); // 现在可以解锁了
		update_size(new_row_bytes); // 更新内存表大小，原子的
	}

	kv_scanner::pointer memory_table::get_scanner (kv_scan_context::pointer p_context)
	{
		memory_table_scanner::pointer scanner(new memory_table_scanner(p_context, this->shared_from_this()));
		if(! scanner)
		{
			LOG(ERROR, "memory_table::get_scanner new memory_table_scanner failed, "
			    << "maybe memory not enough");
			return kv_scanner::pointer();
		}

		// 这里加锁的顺序与freeze()中的操作很有关系！当前的实现
		// 只需要对工作内存表加锁，即可保证它不被放到历史内存表
		// 中。而历史内存表的加入操作只有freeze()一个地方；删除
		// 操作已经被组中获取的history_lock的读锁避免了！
		//
		// 这个地方的锁关系比较复杂，需要细心注意！
		//
		// 要记住：组中，调用该函数前，也对history_lock加了读
		// 锁！所以总的来说，加锁顺序是history_lock, m_lock。
		// freeze()中的顺序要与此保持一致！
		//Guard<ReadWriteLock::ReadLock> guard_working_table(m_lock.read_lock());
		boost::shared_lock<boost::shared_mutex> share_working_tablet(m_lock);

		// 这里要求指明创建的顺序号，所以先遍历添加历史内存表
		// （从前到后，即从老到新），最后添加当前内存表
		size_t n = 0;
		BOOST_FOREACH(const impl_map_pointer &elem, m_history_table)
		{
			scanner->add_history_table(elem, n);
			++n;
		}
		scanner->add_working_table(m_table, n);
		return scanner;
	}

	sstable::pointer memory_table::compact(const uint64_t p_seqnum)
	{
		freeze();
		minor_compact_scanner::pointer scanner = get_compact_scanner();
		sstable::pointer _sstable = compact(scanner, p_seqnum);
		melt();
		return _sstable;
	}

	void memory_table::freeze () {
		// 生成新的内存表
		impl_map_pointer _table(new impl_map_type());

		// 同时对两个数据结构加写锁，避免很多内存表被悬空的状
		// 态，这样实现便得其它操作可以假设工作内存表，要么没被
		// 换掉，要么就已经被换掉并且放在了历史表中！
		//
		// 这一点很重要，并与组中get_scanner的实现很有关系
		//
		// 加锁的顺序对应于：组中对history_lock的读锁、get_scanner中对m_lock的读锁
		//Guard<ReadWriteLock::WriteLock> _guard2(m_history_lock.write_lock());
		//Guard<ReadWriteLock::WriteLock> _guard1(m_lock.write_lock());
		boost::unique_lock<boost::shared_mutex> lock2(m_history_lock);
		boost::unique_lock<boost::shared_mutex> lock1(m_lock);

		// 替换当前的内存表
		{
			// at here, update last compact time
			reset_compact_time(cur_seconds());
			// and update size
			reset_size(0);
			boost::swap(this->m_table, _table);
		}

		// 将原内存表放入到历史列表中
		{
			this->m_history_table.push_back(_table);
		}
	}

	// 去掉了分别对每个列族进行扫描的逻辑，也即去掉了过滤内存表中无用数据的逻辑：
	//
	// 因为内存表扫描器不检查参数，只是简单的返回下一条记录，因此，
	// 若对一个内存表开两个scanner，将导致相应的记录会重复读取两次。
	//
	// 另外，内存表只是一个缓冲，每隔很短时间就会写到sstable，因为
	// 它里面的无用数据必然很少。因此，不过滤也无所谓。
	// 
	minor_compact_scanner::pointer memory_table::get_compact_scanner()
	{
		minor_compact_scanner::pointer mc_scanner(new minor_compact_scanner());
		
		std::vector<std::string> columns(1);
		//BOOST_FOREACH(const family_map_type::value_type &elem, m_family_list)
		{
			//columns[0] = elem.first + ":*"; // only cur family
			columns[0] = "*"; // all column
			// 根据dead_time得出最老时间戳
			// const uint64_t oldest_ts = (elem.second.m_dead_time == 0
			// 			    ? NO_TIMESTAMP
			// 			    : uint64_t(elem.second.m_dead_time) * 1000ULL * 1000ULL);
			const uint64_t oldest_ts = NO_TIMESTAMP;
			kv_scan_context::pointer context(new kv_scan_context(m_tablet_name,
									     m_start_row,
									     m_end_row,
									     columns,
									     NO_TIMESTAMP, // latest_ts
									     oldest_ts,
									     0, // elem.second.m_max_ttl,
									     0, // elem.second.m_max_num_version,
									     (kv_scan_context::OPTION_DISABLE_FILTER_DELETED_ITSELF), // options: filter deleted and not filter deletion itself
									     1)); // num row
			memory_table_scanner::pointer scanner(new memory_table_scanner(context, shared_from_this()));
			//Guard<ReadWriteLock::ReadLock> _guard(m_history_lock.read_lock());
			boost::shared_lock<boost::shared_mutex> _guard(m_history_lock);
			assert(m_compacting_history_table.empty());
			m_compacting_history_table.swap(m_history_table);
			_guard.unlock();
			
			size_t n = 0;
			BOOST_FOREACH(const impl_map_pointer &elem, m_compacting_history_table)
			{
				scanner->add_history_table(elem, n);
				++n;
			}
			LOG(INFO, n << " history tables to be compacted");
			mc_scanner->add(scanner);
		}
		return mc_scanner;
	}

	sstable::pointer memory_table::compact (minor_compact_scanner::pointer scanner,
						const uint64_t p_seqnum)
	{
		if(scanner->eos())
		{
			return sstable::pointer();
		}
		
		LOG_ENV(time_t start_time = cur_seconds());
		
		const std::string _tablet_path = naming_strategy::tablet_path(global::the_conf->xbase_root_path,
									      m_table_name,
									      m_tablet_id,
									      m_start_row,
									      m_end_row);
		const std::string _group_path = naming_strategy::group_path(_tablet_path, m_group_name);

		sstable::pointer _sstable = sstable::create(naming_strategy::sstable_path(_group_path, p_seqnum),
							    naming_strategy::compact_sstable_path(_group_path, p_seqnum),
							    naming_strategy::deleted_sstable_path(_group_path, p_seqnum),
							    m_start_row,
							    m_end_row,
							    p_seqnum,
							    m_group_info.m_block_size,
							    static_cast<compressor::compressor_code>(m_group_info.m_compress_method));
		if(! _sstable)
		{
			LOG(ERROR, "memory_table::compact create sstable with seq num "
			    << p_seqnum << " in group path " << _group_path
			    << " failed, abort");
			// TODO:!!!添加退出程序的机制
			// server.abort();
			assert(false);
			return sstable::pointer();
		}

		LOG_ENV(size_t _count = 0);
		for(; ! scanner->eos(); scanner->next())
		{
			const block_entry &entry = scanner->get();
			const bool ok = _sstable->append(entry);
			LOG_ENV(++_count);
			if (! ok)
			{
				LOG(ERROR, "memory_table::compact append "
				    << _count << "th record ("
				    << dump_string_char(*entry.m_key->row()) << ", "
				    << dump_string_char(*entry.m_key->column()) << ", "
				    << entry.m_key->timestamp()->get() << ", "
				    << dump_string_char(*entry.m_value) 
				    << ") into sstable failed, path = " << _sstable->path());
				// TODO:!!!添加退出程序的机制
				// server.abort();
				assert(false);
				return sstable::pointer();
			}
			else
			{
				LOG(DETAIL, "memory_table::compact append "
					    << _count << "th record ("
					    << dump_string_char(*entry.m_key->row()) << ", "
				    	    << dump_string_char(*entry.m_key->column()) << ", "
				   	    << entry.m_key->timestamp()->get() << ", "
				    	    << dump_string_char(*entry.m_value) 
				   	    << ") into sstable " << _sstable->path());
			}
		}

		const bool ok = _sstable->append_eof();
		if(! ok)
		{
			LOG(ERROR, "memory_table::compact append eof error in sstable "
			    << _sstable->path());
			// TODO:!!!添加退出程序的机制
			// server.abort();
			assert(false);
			return sstable::pointer();
		}

		LOG(INFO, "memory_table::compact " << _count << " records writen to sstable "
		    << _sstable->path() << ", totally used "
		    << cur_seconds() - start_time << " seconds");

		return _sstable;
	}

	// // 重要！
	// // 该操作可能被获取组Scanner的操作阻塞，通过获取到m_history_lock的读锁
	// // 该操作阻塞将阻止返回新创建的SSTable，进而阻止将其加入到组中。
	// void memory_table::remove (const impl_map_pointer &p) {
	// 	Guard<ReadWriteLock::WriteLock> _guard(m_history_lock.write_lock());
	// 	impl_map_list::iterator iter = std::find(m_history_table.begin(),
	// 						m_history_table.end(),
	// 						p);
	// 	if(iter != m_history_table.end())
	// 	{
	// 		this->m_history_table.erase(iter);
	// 	}
	// 	else
	// 	{
	// 		LOG(ERROR, "memory_table::remove not found");
	// 	}
	// }

	void memory_table::melt() {
		//Guard<ReadWriteLock::WriteLock> _guard(m_history_lock.write_lock());
		boost::unique_lock<boost::shared_mutex> _guard(m_history_lock);
		LOG(INFO, m_compacting_history_table.size() << " history tables compacted");
		m_compacting_history_table.clear();
	}

	// DONE: 如何合并多个内存表中的结果？
	// 当且仅当所有内存表中的查询结果都是未删除，结果才是未删除。
	//
	// 传入的内存表必须是已经加锁的
	// bool memory_table::is_deleted (impl_map_pointer p_table,
	// 			       const Key &p_key) const
	// {
	//  	bool deleted = false;
	//  	key_equal_compare_without_timestamp is_key_same;

	//  	if (p_key.timestamp() != NO_TIMESTAMP)
	// 	{
	//  		// 找到第一个比该Key大的（旧的）记录，向前遍历行/列相同
	// 		// 的记录（新的记录），看其值是否为空。
	// 		// 
	// 		// 用反向迭代器从 upper_bound 前面的位置开始遍历
	// 		impl_map_type::reverse_iterator pos(p_table->upper_bound(p_key));
	// 		// 此时 *pos 取到的就是 upper_bound 前面位置的元素
	// 		for (; (pos != p_table->rend()) &&
	// 			     is_key_same(pos->first, p_key);
	// 		     ++pos)
	// 		{
	// 			// 只要比它新的记录有一个是值为空，它就是被删除了的
	// 			if (pos->second.size() == 0)
	// 			{
	// 				deleted = true;
	// 				break;
	// 			}
	// 		}
	// 	}
	// 	else
	// 	{
	// 		Key key(p_key);
	// 		key.time_stamp = cur_micro_seconds(); // 现在 key 是同一行/列中最小的
	// 		impl_map_type::iterator pos(p_table->upper_bound(key)); // 找到比它大的第一个记录
	// 		if ((pos != p_table->end()) && is_key_same(pos->first, key)) // 若行/列相同，它就是该行列现在最新的记录
	// 		{
	// 			if (pos->second.size() == 0) // 现在只需判断最新记录的值是否为空
	// 			{
	// 				deleted = true;
	// 			}
	// 		}
	// 	}

	// 	return deleted;
	// }

	// 依次从当前内存表和所有历史内存表中查询，仅当全部结果为假时，
	// 结果才为假。通常未被删除的数据最多，因此读取这些数据却要从每
	// 个内存表中比较，看起来这使得效率降低，但其实在运行过程中，历
	// 史内存表大部分时间是空的，仅当次要紧缩的过程中，才会有一个内
	// 存表（特殊情况下可能会积累为多个），因此总的来说，效率并没有
	// 太大损失。
	//
	// TODO: 考虑以下实现
	// 
	// 当然，将内存表中的删除条目也放到删除缓存中的话，效率是会提高
	// 不少。这在后续的实现中将考虑，因为现在担心内存使用率，因为所
	// 有的SSTable的删除条目可能会积累得较多（但删除缓存中只需要保
	// 存最后的删除即可）。
	// bool memory_table::is_deleted (const Key &p_key)
	// {
		
	// 	bool deleted = false;

	// 	{
	// 		Guard<ReadWriteLock::ReadLock> _guard(m_lock.read_lock());
	// 		deleted = this->is_deleted(m_table, p_key);
	// 	}

	// 	if (deleted == false)
	// 	{
	// 		Guard<ReadWriteLock::ReadLock> _guard(m_history_lock.read_lock());
	// 		for (impl_map_type_list::iterator iter = m_history_table.begin();
	// 		     iter != m_history_table.end(); ++iter)
	// 		{
	// 			deleted = this->is_deleted(*iter, p_key);
	// 			if (deleted)
	// 				break;
	// 		}
	// 	}

	// 	return deleted;
	// }

	// table_schema::pointer memory_table::get_schema() const
	// {
	// 	if (m_schema.expired())
	// 	{
	// 		Guard<FastLock> _guard(m_schema_lock);
	// 		if (m_schema.expired())
	// 		{
	// 			m_schema = global::the_schema_manager->get_schema(m_table_name);
	// 		}
	// 	}
	// 	return m_schema.lock();
	// }

	// table_schema::group_info::pointer memory_table::get_group_info() const
	// {
	// 	if (m_group_info.expired())
	// 	{
	// 		table_schema::pointer _schema = this->get_schema();
	// 		if (! _schema)
	// 		{
	// 			LOG(ERROR, "memory_table::get_group_info table schema of "
	// 			    << this->m_table_name << " is null.");
	// 		}
	// 		else
	// 		{
	// 			Guard<FastLock> _guard(m_group_info_lock);
	// 			if (m_group_info.expired())
	// 			{
	// 				m_group_info = _schema->get_group_info(m_group_name);
	// 			}
	// 		}
	// 	}
	// 	return m_group_info.lock();
	// }

	std::string memory_table::to_string () const {
		std::ostringstream out;
		out << "{m_size:" << atomic_read(&m_size)
		    << "; threshold:" << m_threshold
		    << "; history_table_size:" << m_history_table.size()
		    << "}";
		return out.str();
	}

} // namespace xbase
